<?php

require_once 'logger.php';


class    Packet
{
    const REQUEST  = 'REQUEST';
    const RESPONSE = 'RESPONSE';
    
    public $action;    // REQUEST/RESPONSE
    public $service;   // service name 
    public $body;      // packet body text 
    public $packid;    // id of packet 
    public $respid;    // which one this packet response  
    
    public function __construct()
    {
        $this->action = NULL;
        $this->service = NULL;
        $this->body = NULL;
        $this->packid = 0;
        $this->respid = 0;
    }
    
    public function setRequest()
    {
        if (self::REQUEST != $this->action) {
            $this->action = self::REQUEST;
        } 
    }
    public function setResponse()
    {
        if (self::RESPONSE != $this->action) {
            $this->action = self::RESPONSE;
        } 
    }
    public function isRequest()
    {
        return (self::REQUEST == $this->action);
    }
    public function isResponse()
    {
        return (self::RESPONSE == $this->action);
    }
    
    public function isComplete() 
    {
        if (empty($this->action) || empty($this->service) || empty($this->body)) {
            return false;
        }
        
        return true;
    }
    
}    // class Packet

class    PacketID
{
    const   MIN_PACKET_ID = 1;
    const   MAX_PACKET_ID = 65535;
    private static $packet_id_current = self::MAX_PACKET_ID;
    
    public static function next()
    {
        if (self::MAX_PACKET_ID == self::$packet_id_current) {
            self::$packet_id_current = self::MIN_PACKET_ID;
        }
        else {
            self::$packet_id_current += 1;
        } 
        
        return self::$packet_id_current;
    } 
}    // class PacketID 

function    depack($record)
{
    $packet = new Packet();
    
    $field_separator = "\t";
    $kv_separator = ':';
    $fields = split($field_separator, $record);
    foreach ($fields as $field) {
        list($k, $v) = split($kv_separator, $field, 2);
        if (empty($v)) {
            continue;
        }
        switch($k) {
        case 'action' :
            $packet->action = $v;
            break;
        case 'service' :
            $packet->service = $v;
            break;
        case 'packid' :
            $packet->packid = $v;
            break;
        case 'respid' :
            $packet->respid = $v;
            break;
        case 'body' :
            $packet->body = $v;
            break;
        }
    }
    if (!$packet->isComplete()) {
        return false;
    }
    
    return $packet;
}    // function depack()

function    enpack(&$packet)
{
    $packid = $packet->packid;
    if (empty($packid)) {
        $packid = PacketID::next();
    }
    
    $fields = array();
    $fields[] = 'action:' . $packet->action;
    $fields[] = 'service:' . $packet->service;
    $fields[] = 'packid:' . $packid;
    if ($packet->respid > 0) {
        $fields[] = 'respid:' . $packet->respid;
    }
    $fields[] = 'body:' . $packet->body;
    
    $record = join("\t", $fields);
    return array($record, $packid);
}    // function enpack()

///////////////////////////////////////////////////////////////////////

function    makeTCPAddress($host, $port)
{
    $address = sprintf("tcp://%s:%s", $host, $port);
    return $address;
}

class    JuTCPChannel   extends Jujucell
{
    const RECORD_END_MARK = "\n";
    const RECORD_END_MARK_SIZE = 1;
    
    const RECV_BUF_SIZE   = 65536;
    
    private $recv_buf;    # for $this->recv() 
    
    private $context;
    private $jucore;
    
    private $logger;
    
    public function __construct($socket, $address=NULL, $context=NULL, $jucore=NULL)
    {
        $this->logger = getLogger();
        
        if (empty($address)) {
            socket_getpeername($socket, $addr_, $port_);
            $address = sprintf("tcp://%s:%s", $addr_, $port_);
            $this->logger->info(sprintf("channel to %s established.", $address));
        }
        
        parent::__construct($socket, $address);
        
        $this->socket = $socket;
        $this->recv_buf = '';
        
        $this->context = $context;
        $this->jucore = $jucore;

        if ($this->jucore) {
            $this->jucore->addChannel($this);
        }
        if ($this->context) {
            $this->context->scheduler->registerChannel($address, $this);
        }
    }
    
    public function OnRecvRecord($data)
    {
        $this->context->scheduler->dispatch($this, $data);
    }
    public function OnPreSendRecord()
    {
        $this->context->scheduler->OnPreSend($this);
    }
    public function OnPostSendRecord()
    {
        $this->context->scheduler->OnPostSend($this);
    }
    public function OnConnectionLost()
    {
        $this->context->scheduler->OnConnectionLost($this);
    }
    
    public function send($data)
    {
        $send_buf = $data;
        while (strlen($send_buf) > 0) {
            $ret = socket_write($this->socket, $send_buf);
            if (FALSE === $ret) {
                $errno = socket_last_error($this->socket);
                //if ((111 == $errno) || (104 == $errno)) { do something }
                $msg = sprintf('Socket WRITE Error -- [%d] %s', $errno, socket_strerror($errno));
                throw new Exception($msg, $errno);
            }
            $send_buf = substr($send_buf, $ret);
            //printf("SEND size=$ret\n");
        }
    }
    
    public function recv(&$data)
    {
        $pos_scan = strlen($this->recv_buf) - self::RECORD_END_MARK_SIZE;
        if ($pos_scan < 0) {
            $pos_scan = 0;
        }
        while(true) {
            $pos = strpos($this->recv_buf, self::RECORD_END_MARK, $pos_scan);
            if (FALSE === $pos) {
                //$ret = socket_read($this->socket, self::RECV_BUF_SIZE);
                $ret = socket_read($this->socket, self::RECV_BUF_SIZE, PHP_NORMAL_READ);
                if (false === $ret) {
                    $errno = socket_last_error($this->socket);
                    //if ((111 == $errno) || (104 == $errno)) { do something }
                    $msg = sprintf('Socket READ Error -- [%d] %s', $errno, socket_strerror($errno));
                    throw new Exception($msg, $errno);
                }
                
                $pos_scan = strlen($this->recv_buf) - self::RECORD_END_MARK_SIZE;
                if ($pos_scan < 0) {
                    $pos_scan = 0;
                }
                $this->recv_buf = join('', array($this->recv_buf, $ret));
                //$this->recv_buf = $this->recv_buf . $ret;
            }
            else {
                $data = substr($this->recv_buf, 0, $pos);
                $this->recv_buf = substr($this->recv_buf, $pos + self::RECORD_END_MARK_SIZE );

                break;
            }
        }
        //var_dump($data);
        return strlen($data);
    }
}    // class JuTCPChannel 


class   JuTCPServer extends Jujucell
{
    public  $port;
    
    private $context;
    private $jucore;
    
    private $logger;
    
    public function __construct($port, $context=None, $jucore=None)
    {
        $this->logger = getLogger();
        
        $sock = socket_create_listen($port);
        if (false === $sock) {
            $errno = socket_last_error();
            $msg = sprintf("Error Creating Listen Socket on port %s -- [%d] %s", $port, $errno, socket_strerror($errno));
            throw new Exception($msg, $errno);
        }
        
        socket_getsockname($sock, $addr_, $port_);
        $address = sprintf("tcp://%s:%s", $addr_, $port_);
        $this->logger->info(sprintf('listening on %s ...', $address));
        parent::__construct($sock, $address);
        
        $this->port = $port;
        $this->context = $context;
        $this->jucore = $jucore;
        
        if ($this->jucore) {
            $this->jucore->addChannel($this);
        }
    }
    
    public function handleReadEvent()
    {
        $client_sock = socket_accept($this->socket);
        socket_getpeername($client_sock, $addr_, $port_);
        $client_address = sprintf("tcp://%s:%s", $addr_, $port_);
        $this->logger->info(sprintf("client %s connected.", $client_address));
        $channel = new JuTCPChannel($client_sock, $client_address, $this->context, $this->jucore);
    }
    
}   // class JuTCPServer

///////////////////////////////////////////////////////////////////////
//

class    ExitException extends Exception
{
}    // class ExitException

class    NotImplementedError extends Exception
{
}    // class NotImplementedError

class    RuntimeError extends Exception
{
}    // class RuntimeError


class    Jujucell
{
    const RECORD_END_MARK = "\n";
    const RECORD_END_MARK_SIZE = 1;
    
    const RECV_BUF_SIZE   = 65536;

    public  $socket;
    public  $address;
    private $connected;
    
    private $in_buf;    // for handleReadEvent()
    private $pos_scan;  
    
    private $out_buf;
    private $producer_fifo;
    
    private $last_error;

    private $logger;
    
    public function __construct($socket, $address)
    {
        $this->socket = $socket;
        $this->connected = true;
        $this->address = $address;
        
        $this->in_buf = '';
        $this->pos_scan = 0;
        
        $this->out_buf = '';
        $this->producer_fifo = new Fifo();
        
        $this->logger = getLogger();
    }
    
    public function OnRecvRecord($data)
    {
        throw new NotImplementedError();
    }
    public function OnPreSendRecord()
    {
        throw new NotImplementedError();
    }
    public function OnPostSendRecord()
    {
        throw new NotImplementedError();
    }
    public function OnConnectionLost()
    {
        throw new NotImplementedError();
    }
    
    
    public function push($data)
    {
        if (empty($data)) {
            throw new Exception('Empty Data to be pushed.');
        }
        
        $producer = new SimpleProducer($data);
        $this->producer_fifo->push($producer);
    }
    
    public function countProducers()
    {
        return $this->producer_fifo->count();
    }
    
    ///////////////////////////////////////////////////////////////////
    
    public function handleReadEvent()
    {
        //$ret = socket_read($this->socket, self::RECV_BUF_SIZE);
        $ret = @socket_read($this->socket, self::RECV_BUF_SIZE, PHP_NORMAL_READ);
        if (false === $ret) {
            $this->logger->debug(sprintf('%s address:%s', __FUNCTION__, $this->address), __FILE__, __LINE__);
            $errno = socket_last_error($this->socket);
            //if ((111 == $errno) || (104 == $errno)) { do something }
            $msg = sprintf('Socket %s READ Error -- [%d] %s', $this->address, $errno, socket_strerror($errno));
            $this->logger->error($msg, __FILE__, __LINE__);
            throw new ExitException($msg, $errno);
        }
        elseif (empty($ret)) {
            $msg = sprintf('Socket %s READ Broke', $this->address);
            throw new ExitException($msg);
        }
        
        //$this->in_buf = join('', array($this->in_buf, $ret));
        $this->in_buf = $this->in_buf . $ret;
        
        while(true) {
            $pos = strpos($this->in_buf, self::RECORD_END_MARK, $this->pos_scan);
            if (FALSE === $pos) {
                $this->pos_scan = strlen($this->in_buf) - self::RECORD_END_MARK_SIZE;
                if ($this->pos_scan < 1) {
                    $this->pos_scan = 0;
                }
                break;
            }
            else {
                $data = substr($this->in_buf, 0, $pos);
                $this->in_buf = substr($this->in_buf, $pos + self::RECORD_END_MARK_SIZE );
                $this->pos_scan = 0;    // scan from head, next time

                // callback data-handler
                $this->OnRecvRecord($data);
            }
        }
    }   // function handleReadEvent()
    
    public function handleWriteEvent()
    {
        $this->sendData();
    }
    
    public function handleExceptionEvent()
    {
        echo __FUNCTION__,__FILE__, __LINE__,"\n";
    }
    public function handleExitException()
    {
        $this->connected = false;
        $this->OnConnectionLost();
    }
    
    public function handleError()
    {
    }
    
    public function readable()
    {
        if ($this->connected) {
            return true;
        }
        else {
            return false;
        }
    }
    
    public function writable()
    {
        if ($this->connected) {
            if ($this->producer_fifo->count() > 0) {
                return true;
            }
        }
        return false;
    }
    
    ///////////////////////////////////////////////////////////////////
    
    private function sendData()
    {
        if (empty($this->out_buf)) {
            $this->refillOutBuffer();
            if (empty($this->out_buf)) {
                return false;
            }
        }
        
        $ret = socket_write($this->socket, $this->out_buf);
        if (FALSE === $ret) {
            $errno = socket_last_error($this->socket);
            //if ((111 == $errno) || (104 == $errno)) { do something }
            $msg = sprintf('Socket WRITE Error -- [%d] %s', $errno, socket_strerror($errno));
            throw new Exception($msg, $errno);
        }
        $this->out_buf = substr($this->out_buf, $ret);

        return true;
    }
    
    private function refillOutBuffer()
    {
        while (true) {
            if ($this->producer_fifo->count() < 1) {
                return false;
            }
            
            $producer = $this->producer_fifo->first();
            $data = $producer->more();
            if (empty($data)) {
                $this->producer_fifo->pop();
                $this->OnPostSendRecord();
                continue;
            }
            else {
                $this->OnPreSendRecord();
                $this->out_buf = join('', array($this->out_buf, $data));
                return true;
            }
        }
        
        return false;
    }
    
}    // class Jujucell

class   SimpleProducer
{
    private $data;
    private $buffer_size;
    
    public function __construct($data, $buffer_size=65536) 
    {
        $this->data = $data;
        $this->buffer_size = $buffer_size;
    }
    
    public function more()
    {
        $result = $this->data;
        $this->data = '';
        
        return $result;
    }
}   // class SimpleProducer

class   Fifo
{
    private $list;
    
    public function __construct()
    {
        $this->list = array();
    }
    
    public function count()
    {
        return count($this->list);
    }
    
    public function isEmpty()
    {
        return ($this->count() < 1);
    }
    
    public function first()
    {
        if ($this->isEmpty()) {
            return NULL;
        }
        return  $this->list[0];
    }
    
    public function push($obj)
    {
        array_push($this->list, $obj);
    }
    
    public function pop()
    {
        if ($this->count() > 0) {
            $result = array_shift($this->list);
            return $result;
        }
        else {
            return NULL;
        }
    }
}   // class Fifo

///////////////////////////////////////////////////////////////////////
//

class    Jujucore
{
    private $map;
    
    public function __construct()
    {
        $this->map = array();
    }
    
    public function addChannel($channel)
    {
        $sock = $channel->socket;
        if (isset($this->map[ $sock ])) {
            return true;
        }
        $this->map[ $sock ] = $channel;
    }
    
    public function delChannel($channel)
    {
        $sock = $channel->socket;
        if (isset($this->map[ $sock ])) {
            unset($this->map[ $sock ]);
        }
        return true;
    }
    
    public function loop($timeout=10, $count=0)
    {
        if ($timeout > 0) {
            $tv_sec = floor($timeout);
            $tv_usec = floor(($timeout - $tv_sec) * 1000000);
        }
        else {
            $tv_sec = 0;
            $tv_usec = 0;
        }
        $count_ = 0;
        
        while( count($this->map) > 0 ) {
            $r = array();
            $w = array();
            $e = array();
            
            foreach ($this->map as $sock => $obj) {
                $exceptable = false;
                if ($obj->readable()) {
                    $r[] = $obj->socket;
                    $exceptable = true;
                }
                if ($obj->writable()) {
                    $w[] = $obj->socket;
                    $exceptable = true;
                }
                if ($exceptable) {
                    $e[] = $obj->socket;
                }
            }
            
            $ret = socket_select($r, $w, $e, $tv_sec, $tv_usec);
            if (FALSE === $ret) {
                $errno = socket_last_error();
                $msg = sprintf('Socket SELECT Error -- [%d] %s', $errno, socket_strerror($errno));
                throw new Exception($msg, $errno);
            }
            elseif (0 === $ret) {
                return 0;
            }
            
            // handle exception
            foreach ($e as $sock) {
                $obj = $this->map[ $sock ];
                try {
                    $obj->handleExceptionEvent();
                }
                catch(ExitException $ex) {
                    unset($this->map[ $sock ]);
                }
                catch(Exception $ex) {
                    $obj->handleError();
                }
            }
                        
            // handle reading
            foreach ($r as $sock) {
                $obj = $this->map[ $sock ];
                try {
                    $obj->handleReadEvent();
                }
                catch(ExitException $ex) {
                    $obj->handleExitException();
                    unset($obj);
                    unset($this->map[ $sock ]);
                }
                catch(Exception $ex) {
                    $obj->handleError();
                }
            }
            
            // handle writing            
            foreach ($w as $sock) {
                $obj = $this->map[ $sock ];
                try {
                    $obj->handleWriteEvent();
                }
                catch(ExitException $ex) {
                    unset($this->map[ $sock ]);
                }
                catch(Exception $ex) {
                    $obj->handleError();
                }
            }
            
            // count loop 
            if ($count > 0) {
                ++$count_;
                if ($count_ >= $count) {
                    break;
                }
            }
        } // while(true)
    }    // loop()
    
}    // class Jujucore

///////////////////////////////////////////////////////////////////////

class   Server
{
    public function service($service, $record, $address_from)
    {
        throw new NotImplementedError();
    }
    
}   // class Server

///////////////////////////////////////////////////////////////////////

function    passby($params=NULL)
{
    return $params;
}

class   BaseAction
{
    const  ACTION_TYPE_NONE = 0;
    const  ACTION_TYPE_SEND = 1;
    const  ACTION_TYPE_RECV = 2;
    const  ACTION_TYPE_SENDnRECV = 3;
    
    const  DATA_TYPE_REQUEST  = 10;
    const  DATA_TYPE_RESPONSE = 11;
    
    protected $action_type;
    public  $job;
    
    public  $remote_address;
    public  $remote_service;
    
    public  $data;
    public  $data_type;
    public  $data_respid;
    
    private $callbacks;
    private $result;
    private $reason;
    
    public function __construct($job=NULL)
    {
        $this->action_type = self::ACTION_TYPE_NONE;
        $this->job = $job;
        
        $this->data = '';
        $this->data_type = self::DATA_TYPE_REQUEST;
        $this->data_respid = 0;
        
        $this->callbacks = array();
    }
    
    public function setRespid($respid)
    {
        if ($respid > 0) {
            $this->data_type = self::DATA_TYPE_RESPONSE;
            $this->data_respid = $respid;
        }
        else {
            $this->data_type = self::DATA_TYPE_REQUEST;
            $this->data_respid = 0;
        }
    }
    public function isResponseData()
    {
        return ($this->data_type == self::DATA_TYPE_RESPONSE);
    }
    
    public function isSendAction()
    {
        return (self::ACTION_TYPE_SEND == $this->action_type);
    }
    public function isRecvAction()
    {
        return (self::ACTION_TYPE_RECV == $this->action_type);
    }
    public function isSendRecvAction()
    {
        return (self::ACTION_TYPE_SENDnRECV == $this->action_type);
    }
    
    public function addCallBack($callback)
    {
        $this->addCallbacks($callback, 'passby');
    }
    
    public function addErrorBack($errorback)
    {
        $this->addCallbacks('passby', $errorback);
    }
    
    public function addCallbacks($callback, $errorback)
    {
        $this->callbacks[] = array($callback, $errorback);
    }
    
    public function done($data='')
    {
        $this->result = $data;
        while (count($this->callbacks) > 0) {
            list($obj_callback, $obj_errorback) = array_shift($this->callbacks);
            if (is_array($obj_callback)) {
                $obj = $obj_callback[0];
                $method = $obj_callback[1];
                $this->result = $obj->$method($this->result);
            }
            elseif (is_string($obj_callback)) {
                $callback = $obj_callback;
                $this->result = $callback($this->result);
            }
            else {
                throw new RuntimeError('Invalid Action CallBack "%s"', print_r($obj_callback));
            }
        }
    }
    
    public function fail($reason='')
    {
        $this->result = $data;
        while (count($this->callbacks) > 0) {
            list($obj_callback, $obj_errorback) = array_shift($this->callbacks);
            if (is_array($obj_errorback)) {
                $obj = $obj_errorback[0];
                $method = $obj_errorback[1];
                $this->result = $obj->$method($this->result);
            }
            elseif (is_string($obj_errorback)) {
                $callback = $obj_errorback;
                $this->result = $callback($this->result);
            }
            else {
                throw new RuntimeError('Invalid Action ErrorBack "%s"', print_r($obj_callback));
            }
        }
    }
    
}   // class BaseAction

class   SendAction  extends BaseAction
{
    public function __construct($job=NULL)
    {
        parent::__construct($job);
        $this->action_type = parent::ACTION_TYPE_SEND;
    }
}   // class SendAction

class   RecvAction  extends BaseAction
{
    public function __construct($job=NULL)
    {
        parent::__construct($job);
        $this->action_type = parent::ACTION_TYPE_RECV;
    }
}   // class SendAction

class   SendRecvAction  extends BaseAction
{
    public function __construct($job=NULL)
    {
        parent::__construct($job);
        $this->action_type = parent::ACTION_TYPE_SENDnRECV;
    }
}   // class SendAction

///////////////////////////////////////////////////////////////////////

class   BaseJob
{
    public function currentAction()
    {
        throw new NotImplementedError(__FUNCTION__);
    }
    
    /**
     * @brief return next action
     *
     * if no more action, return NULL
     * if no action at present, but job not end, return '' (scheduler would degrade priority of this job)
     */
    public function nextAction()
    {
        throw new NotImplementedError(__FUNCTION__);
    }
    
}   // class BaseJob

///////////////////////////////////////////////////////////////////////

class   Action4Channel
{
    private $q_send_actions;            // actions want to send some outgoing packets
    private $q_recv_actions;            // actions want to accept any incoming packets
    private $q_wait_response_actions;   // actions waiting for some response
    
    private $q_in_packets;              // packets cannot handle would hold here
    
    public  $sending_packid;
    
    private $logger;
    
    public function __construct()
    {
        $this->q_send_actions = new Fifo();
        $this->q_recv_actions = new Fifo();
        $this->q_wait_response_actions = array();
        $this->q_in_packets = new Fifo();
        $this->sending_packid = 0;
        
        $this->logger = getLogger();
    }
    
    public function addSendAction($action)
    {
        if (empty($action)) {
            $msg = sprintf('%s(): Invalid EMPTY action comes to queue-send.', __FUNCTION__);
            throw new RuntimeError($msg);
        }
        $this->q_send_actions->push($action);
    }
    public function countSendActions()
    {
        return $this->q_send_actions->count();
    }
    public function firstSendAction()
    {
        return $this->q_send_actions->first();
    }
    public function popSendAction()
    {
        if ($this->q_send_actions->count() > 0) {
            $result = $this->q_send_actions->pop();
            return $result;
        }
        return NULL;
    }
    
    public function addRecvAction($action)
    {
        if (empty($action)) {
            $msg = sprintf('%s(): Invalid EMPTY action comes to queue-recv.', __FUNCTION__);
            throw new RuntimeError($msg);
        }
        $this->q_recv_actions->push($action);
    }
    
    public function addWaitResponseAction($action, $respid)
    {
        if (empty($action)) {
            $msg = sprintf('%s(): Invalid EMPTY action comes to queue-response.', __FUNCTION__);
            throw new RuntimeError($msg);
        }
        if (isset($this->q_wait_response_actions[$respid])) {
            $actions = $this->q_wait_response_actions[$respid];
        }
        else {
            $actions = new Fifo();
            $this->q_wait_response_actions[$respid] = $actions;
        }
        
        $actions->push($action);
    }
    
    public function popRecvAction()
    {
        if ($this->q_recv_actions->count() > 0) {
            $result = $this->q_recv_actions->pop();
            return $result;
        }
        return NULL;
    }
    
    public function popRecvActionFor($respid)
    {
        if (isset($this->q_wait_response_actions[$respid])) {
            $actions = $this->q_wait_response_actions[$respid];
            
            if ($actions->count() > 0) {
                $result = $actions->pop();
                
                if ($actions->count() < 1) {
                    unset($this->q_wait_response_actions[$respid]);
                }
                
                return $result;
            }
        }
        
        return NULL;
    }
    
    public function pushIncomePacket($packet)
    {
        if (empty($packet)) {
            $msg = sprintf('%s(): Invalid EMPTY packet comes to queue-income.', __FUNCTION__);
            throw new RuntimeError($msg);
        }
        $this->q_in_packets->push($packet);
    }
    public function popIncomePacket()
    {
        $count = $this->countIncomePackets();
        
        $packet = $this->q_in_packets->pop();

        if (empty($packet) && ($count > 0)) {
            $msg = sprintf('%s(): Invalid EMPTY packet pop off queue-income (count=%d).', __FUNCTION__, $count);
            throw new RuntimeError($msg);
        }
        
        return $packet;
    }
    public function countIncomePackets()
    {
        return $this->q_in_packets->count();
    }
    
}   // class Action4Channel

class   JuScheduler
{
    const   RECORD_END_MARK = "\n";
    
    private $table_services;        // map { service --> server }
    private $table_channels;        // map { address --> channel }
    private $action_list_for_channel;
    
    private $job_list_ready;
    private $job_list_done;
    private $job_list_fail;
    private $job_list_run;
    
    private $logger;
    
    public function __construct()
    {
        $this->table_services = array();
        $this->table_addresses = array();
        $this->action_list_for_channel = array();
        
        $this->job_list_ready = new Fifo();
        $this->job_list_done  = new Fifo();
        $this->job_list_fail  = new Fifo();
        $this->job_list_run   = new Fifo();
        
        $this->logger = getLogger();
    }
    
    public function dispatch($channel_from, $record)
    {
        if ($this->logger->debugable()) {
            if (strlen($record) < 256) {
                $this->logger->debug(sprintf("dispatch (CAHNNEL:%s size:%d) `%s`", $channel_from->address, strlen($record), $record), __FILE__, __LINE__);
            }
            else {
                $this->logger->debug(sprintf("dispatch (CAHNNEL:%s size:%d)", $channel_from->address, strlen($record)), __FILE__, __LINE__);
            }
        }
        
        $packet = depack($record);
        if (empty($packet)) {
            $msg = sprintf('Invalid record : %s', $record);
            throw new RuntimeError($msg);
        }
        
        if ($packet->isRequest()) {
            $service = $packet->service;
            if (empty($service)) {
                throw new RuntimeError('Invalid EMPTY service name.');
            }
            if (!isset($this->table_services[$service])) {
                $msg = sprintf('SERVICE "%s" was NOT registered.', $service);
                throw new RuntimeError($msg);
            }
            $server = $this->table_services[ $service ];
            list($service_to, $response, $address_to) = $server->serve($service, $packet->body, $channel_from->address);
            if (empty($response)) {
                // no response
                return  true;
            }
            $action = new SendAction();
            $action->remote_service = $service_to;
            $action->remote_address = $address_to;
            $action->data = $response;
            $action->setRespid($packet->packid);
            $this->addSendAction($action, $address_to);
            return  true;
        }
        elseif ($packet->isResponse()) {
            $address = $channel_from->address;
            if (isset($this->action_list_for_channel[$address])) {
                $action_list = $this->action_list_for_channel[$address];
                
                $action = $action_list->popRecvActionFor($packet->respid);
                if (empty($action)) {
                    $action_list->pushIncomePacket($packet);
                    return true;
                }
                
                // action callback
                $action->done($packet->body);
                // move job to done-list
                $job = $action->job;
                $this->job_list_done->push($job);
            }
            else {
                $this->logger->error('No action-list for address %s', $address);
                $action_list = new Action4Channel();
                $this->action_list_for_channel[$address] = $action_list;
                
                $action_list->pushIncomePacket($packet);
                return true;
            }
        }
        else {
            throw new NotImplementedError();
        }
    }
    
    ///////////////////////////////////////////////////////////////////
    
    public function registerService($service, $server)
    {
        $this->table_services[ $service ] = $server;
    }
    public function unregisterService($service, $server)
    {
        unset($this->table_services[ $service ]);
    }
    
    ///////////////////////////////////////////////////////////////////

    public function registerChannel($address, $channel)
    {
        $this->table_channels[ $address ] = $channel;
    }
    public function unregisterChannel($address, $channel=NULL)
    {
        if (isset($this->table_channels[$address])) {
            unset($this->table_channels[$address]);
        }
    }
    public function lookupChannel($address)
    {
        //$this->logger->debug(sprintf('%s address:%s %s', __FUNCTION__, $address, print_r($this->table_channels, true)), __FILE__, __LINE__);
        if (isset($this->table_channels[$address])) {
            return $this->table_channels[$address];
        }
        return false;
    }
    
    ///////////////////////////////////////////////////////////////////
    
    private function addSendAction($action, $address)
    {
        $this->logger->debug(sprintf('%s address:%s', __FUNCTION__, $address), __FILE__, __LINE__);
        $channel = $this->lookupChannel($address);
        if (empty($channel)) {
            $msg = sprintf('No channel for address "%s"', $address);
            throw new RuntimeError($msg);
        }
        if (isset($this->action_list_for_channel[$address])) {
            $action_list = $this->action_list_for_channel[$address];
        }
        else {
            $action_list = new Action4Channel();
            $this->action_list_for_channel[$address] = $action_list;
        }
        $action_list->addSendAction($action);
        //$this->logger->debug(sprintf('%s %s countSendActions=%s', __FUNCTION__, $address, $action_list->countSendActions()), __FILE__, __LINE__);
    }
    
    ///////////////////////////////////////////////////////////////////
    
    public function addJob($job)
    {
        $this->job_list_done->push($job);
    }
    
    private function scheduleFailJobs()
    {
        while( $this->job_list_fail->count() > 0 ) {
            $job = $this->job_list_fail->pop();
            $this->logger->error('ToBeImplemented: Job failed on Action.');
            unset( $job );
        }
    }
    
    private function scheduleDoneJobs()
    {
        $count_job = $this->job_list_done->count();
        for ($i=0; $i<$count_job; $i++) {
            $job = $this->job_list_done->pop();
            $action = $job->nextAction();
            if ($action) {
                $this->job_list_ready->push($job);
            }
            elseif (NULL === $action) {
                # job done, no more action
                unset($job);
            }
            else {
                $this->job_list_done->push($job);
            }
        }
    }
    
    private function scheduleReadyJobs()
    {
        $count_job = $this->job_list_ready->count();
        for ($i=0; $i<$count_job; $i++) {
            $job = $this->job_list_ready->pop();
            $current_action = $job->currentAction();
            
            if($current_action->isSendAction()) {
                $address = $current_action->remote_address;
                $this->addSendAction($current_action, $address);
                //$this->job_list_run->push($job);
            }
            elseif ($current_action->isRecvAction()) {
                $address = $current_action->remote_address;
                $channel = $this->lookupChannel($address);
                if (empty($channel)) {
                    $msg = sprintf('No channel for address %s', $address);
                    throw new RuntimeError($msg);
                }
                if (isset($this->action_list_for_channel[$address])) {
                    $action_list = $this->action_list_for_channel[$address];
                }
                else {
                    $action_list = new Action4Channel();
                    $this->action_list_for_channel[$address] = $action_list;
                }
                $action_list->addRecvAction($current_action);
                //$this->job_list_run->push($job);
            }
            elseif ($current_action->isSendRecvAction()) {
                $address = $current_action->remote_address;
                $channel = $this->lookupChannel($address);
                if (empty($channel)) {
                    $msg = sprintf('No channel for address %s', $address);
                    throw new RuntimeError($msg);
                }
                if (isset($this->action_list_for_channel[$address])) {
                    $action_list = $this->action_list_for_channel[$address];
                }
                else {
                    $action_list = new Action4Channel();
                    $this->action_list_for_channel[$address] = $action_list;
                }
                $action_list->addSendAction($current_action);
                //$this->job_list_run->push($job);
            }
            else {
                list($ok, $crump) = $job->execute();
                if ($ok) {
                    $job->done($crump);
                    $this->job_list_done->push($job);
                }
                else {
                    $job->fail($crump);
                    $this->job_list_fail->push($job);
                }
            }
        }
    }
    
    private function scheduleConsumers()
    {
        foreach ($this->action_list_for_channel as $address => $action_list) {
            //$this->logger->debug(sprintf('%s() address:%s', __FUNCTION__, $address), __FILE__, __LINE__);
            $count_packets = $action_list->countIncomePackets();
            for ($i=0; $i<$count_packets; $i++) {
                $packet = $action_list->popIncomePacket();
                if (empty($packet)) {
                    // why is the packet empty?
                    $this->logger->error(sprintf('EMPTY packet:%s, %d packets still in queue-income.', print_r($packet, true), $action_list->countIncomePackets()), __FILE__, __LINE__);
                    continue; 
                }
                
                if ($packet->isResponse()) {
                    $action = $action_list->popRecvActionFor($packet->respid);
                    if (empty($action)) {
                        // NOT found, hold it again
                        $action_list->pushIncomePacket($packet);
                        continue;
                    }
                    
                    // action callback
                    $action->done($packet->body);
                    // move action to done-list
                    $job = $action->job;
                    $this->job_list_done->push($job);
                }
                else {
                    // hold it again
                    $action_list->pushIncomePacket($packet);
                }
            }
        }
    }
    
    private function scheduleProducers()
    {
        foreach ($this->action_list_for_channel as $address => $action_list) {
            //$this->logger->debug(sprintf('%s() address:%s', __FUNCTION__, $address), __FILE__, __LINE__);
            if ($action_list->countSendActions() < 1) {
                continue;
            }
            $channel = $this->lookupChannel($address);
            if (empty($channel)) {
                $msg = sprintf('No channel for address "%s"', $address);
                throw new RuntimeError($msg);
            }
            if ($channel->countProducers() > 0) {
                continue;
            }
            
            $action = $action_list->firstSendAction();
            $packet = new Packet();
            if ($action->isResponseData()) {
                $packet->respid = $action->data_respid;
                $packet->setResponse();
            }
            else {
                $packet->setRequest();
            }
            $packet->body = $action->data;
            $packet->service = $action->remote_service;
            //$this->logger->debug(sprintf('action:%s', print_r($action, true)), __FILE__, __LINE__);
            //$this->logger->debug(sprintf('packet:%s', print_r($packet, true)), __FILE__, __LINE__);
            list($record, $packid) = enpack($packet);
            if ($this->logger->debugable()) {
                if (strlen($record) < 256) {
                    $this->logger->debug(sprintf('CHANNEL %s PUSH (packid=%d size=%d)%s', $address, $packid, strlen($record), $record), __FILE__, __LINE__);
                }
                else {
                    $this->logger->debug(sprintf('CHANNEL %s PUSH (packid=%d size=%d)', $address, $packid, strlen($record)), __FILE__, __LINE__);
                }
            }
            $action_list->sending_packid = $packid;
            $channel->push($record . self::RECORD_END_MARK);
        }
    }

    public function OnPreSend($channel)
    {
    }
    public function OnPostSend($channel)
    {
        $address = $channel->address;
        if (!isset($this->action_list_for_channel[$address])) {
            $msg = sprintf('Suspicious channel address "%s"', $address);
            throw new RuntimeError($msg);
        }
        $actions = $this->action_list_for_channel[$address];
        $action = $actions->popSendAction();
        $job = $action->job;
        if (empty($job)) {
            // a standalone action
            return true;
        }
        
        if ($action->isSendAction()) {
            //schedule job to done-list
            $this->job_list_done->push( $job );
        }
        elseif ($action->isSendRecvAction()) {
            // schedule action to recv Queue
            // register respid to wait
            //log.debug('CHANNEL %s sending packet (id:%d)', channel, actions.sending_packid)
            //log.debug('Action RecvResponse for %d', actions.sending_packid )
            $actions->addWaitResponseAction($action, $actions->sending_packid);
            $actions->sending_packid = 0;
        }
        else {
            // suspicious action
        }
    }
    
    public function OnConnectionLost($channel)
    {
        $address = $channel->address;
        $this->logger->debug(sprintf('%s(%s)', __FUNCTION__, $address), __FILE__, __LINE__);
        if (isset($this->action_list_for_channel[$address])) {
            unset($this->action_list_for_channel[$address]);
        }
        $this->unregisterChannel($address);
    }
    
    
    ///////////////////////////////////////////////////////////////////
    
    public function loop_once()
    {
        $this->scheduleFailJobs();
        $this->scheduleDoneJobs();
        $this->scheduleReadyJobs();
        
        $this->scheduleConsumers();
        $this->scheduleProducers();
    }

}   // class JuScheduler

class   JuContext
{
    public $scheduler;
    
    public function __construct($scheduler)
    {
        $this->scheduler = $scheduler;
    }
    
}   // class JuContext

///////////////////////////////////////////////////////////////////////

class   Jujube
{
    private $jucore;
    private $context;
    
    private $listeners;
    private $logger;
    
    public function __construct()
    {
        $scheduler = new JuScheduler();
        $context = new JuContext($scheduler);
        $this->context = $context;
        
        $jucore = new Jujucore();
        $this->jucore = $jucore;
        
        $this->listeners = array();
        
        $this->logger = getLogger();
    }
    
    public function listenTCP($port)
    {
        if (isset($this->listeners[$port])) {
            return  true;
        }
        
        $listener = new JuTCPServer($port, $this->context, $this->jucore);
        $this->listeners[ $port ] = $listener;
    }
    
    public function connectTCP($host, $port)
    {
        $sock = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
        $ok = socket_connect($sock, $host, $port);
        if (false === $ok) {
            $errno = socket_last_error($sock);
            $msg = sprintf("Error Connecting Socket to %s:%s -- [%d] %s", $host, $port, $errno, socket_strerror($errno));
            throw new Exception($msg, $errno);
        }
        $address = makeTCPAddress($host, $port);
        $channel = new JuTCPChannel($sock, $address, $this->context, $this->jucore);
    }
    
    public function registerService($service, $server)
    {
        $this->context->scheduler->registerService($service, $server);
    }
    
    public function unregisterService($service, $server)
    {
        $this->context->scheduler->unregisterService($service, $server);
    }
    
    public function addJob($job)
    {
        $this->context->scheduler->addJob($job);
    }
    
    public function run()
    {
        $timeout = 0.010;
        $loop_count = 1;
        
        while(true) {
            try {
                $this->jucore->loop($timeout, $loop_count);
                $this->context->scheduler->loop_once();
            }
            catch(Exception $ex) {
                $this->logger->error($ex->getMessage());
                throw $ex;
            }
        }
    }
    
}   // class Jujube

